#include <stdlib.h>
#include <errno.h>
#include <signal.h>
#include "threadpool.h"
#include "platform.h"


void init_thread_pool(ThreadPool *pool)
{

    initQueue(&pool->queue_);
    pthread_mutex_init(&pool->mu_, NULL);
    pthread_cond_init(&pool->cond_, NULL);

    for (int i = 0; i < MAXTHREAD; i++)
    {
        ClerkThread_t *thread = &pool->clerk_threads[i];
        thread->thread_state_ = FREE;
        thread->pool_mu_ = &pool->mu_;
        thread->pool_cond_ = &pool->cond_;
        thread->pool_queue_ = &pool->queue_;
        pipe(thread->thread_pipe_);
    }

    for (int i = 0; i < MAXWORKER; i++)
    {
        WorkerThread_t *worker = &pool->workers_[i];
        worker->pool_mu_ = &pool->mu_;
        worker->pool_cond_ = &pool->cond_;
        worker->pool_queue_ = &pool->queue_;
        pthread_create(&worker->thread_indentider_, NULL, run_worker, (void *)worker);
    }

    printf("init thread pool success\n");
    fflush(stdout);
}

void join_thread(ClerkThread_t *thread)
{
    if (thread->thread_state_ != RUNNING) {
        return;
    }
    int sig = SIGINT;
    write(thread->thread_pipe_[0], &sig, sizeof(sig));

    for (int i = 0; i < thread->numfd_; i++)
    {
        Reply rep = {0, QUIT};
        send(thread->thread_fds_[i], &rep, MAXSIZE, 0);
        close(thread->thread_fds_[i]);
    }

    thread->numfd_ = 0;
    thread->thread_state_ = FREE;
    close(thread->thread_pipe_[0]);
    close(thread->thread_pipe_[1]);
    pthread_join(thread->thread_identifer_, NULL);
}

void close_thread_pool(ThreadPool *pool)
{
    for (int i = 0; i < MAXTHREAD; i++) {
        join_thread(&pool->clerk_threads[i]);
    } 
    for (int i = 0; i < MAXWORKER; i++) {
        pthread_kill(pool->workers_[i].thread_indentider_, SIGINT);
        pthread_join(pool->workers_[i].thread_indentider_, NULL);
    }
}

int get_fd_set(const ClerkThread_t *thread, fd_set *readset)
{
    FD_ZERO(readset);
    FD_SET(thread->thread_pipe_[0], readset);

    int numfd = thread->numfd_;
    int maxfd = 0;

    for (int i = 0; i < numfd; i++)
    {
        int fd = thread->thread_fds_[i];
        maxfd = maxfd > fd ? maxfd : fd;
        FD_SET(fd, readset);
    }

    return maxfd;
}   

void run_thread(ClerkThread_t *thread)
{
    if (thread->thread_state_ != FREE)
    {
        return;
    }
    thread->thread_state_ = RUNNING;
    pthread_create(&thread->thread_identifer_, NULL, run_clerk, (void *)thread);
}

int add_fd(ThreadPool *pool, int fd)
{
    int min_num_fd = MAXFDPERTHREAD;
    int min_thread_id;

    for (int i = 0; i < MAXTHREAD; i++)
    {
        ClerkThread_t *thread = &pool->clerk_threads[i];
        if (thread->thread_state_ == FREE) {
            run_thread(thread);
            thread->thread_fds_[0] = fd;
            thread->numfd_ = 1;
            return 0;
        }

        if (thread->thread_fds_[i] < min_num_fd) {
            min_num_fd = thread->thread_fds_[i];
            min_thread_id = i;
        }
    }

    if (min_num_fd == MAXFDPERTHREAD) {
        return -1;
    }

    ClerkThread_t *thread = &pool->clerk_threads[min_thread_id];
    thread->thread_fds_[thread->numfd_++] = fd;

    return 0;
}

int delete_fd_in_thread(ClerkThread_t *thread, int fd) {

    if (thread->thread_state_ != RUNNING) {
        return -1;
    }

    for (int j = 0; j < thread->numfd_; j++)
    {
        if (thread->thread_fds_[j] == fd) {
            thread->thread_fds_[j] = thread->thread_fds_[thread->numfd_ - 1];
            if (--thread->numfd_ == 0) {
                join_thread(thread);
            }
            return 1;
        }
    }
    
    return -1;
}

int delete_fd(ThreadPool *pool, int fd) {

    for (int i = 0; i < MAXTHREAD; i++)
    {
        ClerkThread_t *thread = &pool->clerk_threads[i];

        if (delete_fd_in_thread(thread, fd)) {
            return 1;
        }
    }

    return -1;
}

void *run_clerk(void *thread_args)
{
    ClerkThread_t *thread = (ClerkThread_t *)thread_args;
    fd_set readset;
    int *pipe = thread->thread_pipe_;
    int *fds = thread->thread_fds_;
    pthread_mutex_t *mu = thread->pool_mu_;
    pthread_cond_t *cond = thread->pool_cond_;
    ArrayQueue *queue = thread->pool_queue_;

    while(1)
    {
        int max_fd = get_fd_set(thread, &readset);
        int ready = select(max_fd + 1, &readset, NULL, NULL, NULL);

        if (ready == -1 && errno == EINTR) {
            continue;
        }
        if (FD_ISSET(pipe[0], &readset)) {
            int sig;
            read(pipe[1], &sig, sizeof(sig));
            if (sig == SIGINT) {
                break;
            }
        }
        int fds_to_delete[MAXFDPERTHREAD];
        int num_fd_to_delete = 0;;

        for (int i = 0; i < thread->numfd_; i++)
        {
            if (FD_ISSET(fds[i], &readset)) {
                Command cmd;
                ssize_t size;
                if ((size = recv(fds[i], &cmd, sizeof(Command), 0)) > 0) {
                    if (cmd.type == QUIT) {
                        fds_to_delete[num_fd_to_delete++] = fds[i];
                    } 
                    Request req = {fds[i], cmd};
                    pthread_mutex_lock(mu);
                    enqueue(queue, req);
                    pthread_cond_signal(cond);
                    pthread_mutex_unlock(mu);
                } else {
                    fds_to_delete[num_fd_to_delete++] = fds[i];
                }
            }
        }

        for (int i = 0; i < num_fd_to_delete; i++)
        {
            delete_fd_in_thread(thread, fds_to_delete[i]);
        }
    }
    return NULL;
}


void *run_worker(void *worker_args)
{

    WorkerThread_t *worker = (WorkerThread_t *)worker_args;
    pthread_mutex_t *mu = worker->pool_mu_;
    pthread_cond_t *cond = worker->pool_cond_;
    ArrayQueue *queue = worker->pool_queue_;
    struct timespec timeout;

    while (1) 
    { 
        pthread_mutex_lock(mu);
        clock_gettime(CLOCK_REALTIME, &timeout);
        timeout.tv_sec += 1;

        while(isEmpty(queue))
        {
            pthread_cond_wait(cond, mu);
        }

        Request req;
        dequeue(queue, &req);

        pthread_mutex_unlock(mu);

        Command cmd = req.cmd;
        Reply reply = execute_cmd(&cmd);
        send(req.client_fd, &reply, MAXSIZE, 0);
    }
}